Skip to content

feat(parquet): add bytes_processed scan-completion metric - #24524

Merged
adriangb merged 6 commits into
apache:mainfrom
pydantic:claude/query-progress-tracker-metrics-dyl2u1
Aug 21, 2026
Merged

feat(parquet): add bytes_processed scan-completion metric#24524
adriangb merged 6 commits into
apache:mainfrom
pydantic:claude/query-progress-tracker-metrics-dyl2u1

Conversation

@adriangb

@adriangb adriangb commented Aug 20, 2026

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

  • No issue filed yet. Happy to open one describing the problem before review if that is preferred.

Rationale for this change

Nothing today reports how far along a scan is. That is what a progress bar, a query-progress API, or a watchdog that cancels runaway queries needs, and both existing signals fail at it in opposite directions:

  • bytes_scanned looks like the right metric, because its natural denominator — the size of the files in the plan — is known before the query runs. But it counts only the bytes actually fetched. A scan that prunes most of its row groups, or projects 3 of 50 columns, reports a small fraction of the file even once it has finished, so bytes_scanned / total file bytes understates progress by a factor that varies per query and is not knowable up front.
  • files_processed only moves when a whole file completes. With high scan parallelism it reads 0% through the entire first wave of files, however much work is in flight, and then jumps.

The gap is not the unit, it is the numerator: a scan is done with a byte once it has either read it or proved it does not need it, and bytes_scanned counts only the first half.

What changes are included in this PR?

A new bytes_processed counter on ParquetFileMetrics, alongside bytes_scanned and sharing its metric type, category and per-file label:

bytes_scanned    — bytes fetched from the object store
bytes_processed  — bytes the scan is finished with, whether read or pruned

Its contract is one invariant:

Over the lifetime of a file — or, for a file split into byte ranges for parallelism, a range — bytes_processed advances by exactly effective_size(), monotonically.

so bytes_processed / total file bytes is a completion fraction that needs no statistics and no second metric to normalise against.

Credit lands a row group at a time:

event credit
file/range pruned before open (file statistics, dynamic filter) effective_size()
row groups the final access plan skips — range, statistics, bloom filter, limit, page index compressed_size(rg), at open
row group dropped mid-scan by a dynamic filter compressed_size(rg)
row group reached by the decoder compressed_size(rg)
file closed for any reason (finished, LIMIT, early stop, error) the remainder

Two details worth a reviewer's attention:

  • ByteProgress (in metrics.rs) clamps every credit to the bytes left in the range and credits the remainder on Drop. The clamp absorbs the two inexactnesses of crediting by row group — a file is slightly larger than the sum of its row groups, and a row group is assigned to a byte range by its first page's offset — and Drop is what makes the invariant hold under LIMIT, EarlyStoppingStream and errors without instrumenting every exit path.
  • row_group_in_range is extracted out of RowGroupAccessPlanFilter::prune_by_range so the open-time skip pass decides which row groups a range owns by the same rule, instead of duplicating the offset logic. Without it, one range of a split file would credit its siblings' row groups against its own budget and jump straight to 100% at open.

Folding every open-time pruning stage into a single pass over the final access plan is what keeps this free of double counting: range, statistics, bloom, limit and page-index pruning have all been applied by then, so there is no need to instrument five sites and reason about their overlap.

Cost is one atomic add per row group. Crediting a row group's bytes progressively as its rows decode — proportionally by rows resolved, trued up at the boundary — is deliberately left to a follow-up; it does not change this metric's contract.

Are these changes tested?

Yes. Seven new tests in opener::test::bytes_processed, all asserting the invariant, for: a plain scan, a scan pruning row groups, a file pruned before open by a dynamic filter (also asserting bytes_scanned == 0), a byte-split file (each range credits exactly its own length, and the ranges sum to the file), a LIMIT that ends the scan early, a file that errors before it can be opened at all, and one that steps through the stream batch by batch to pin that credit advances during the scan rather than all at once on close.

The two carrying the real claims were mutation-checked: removing the mid-scan credit fails credit_advances_while_the_scan_runs, and removing the open-time credit fails row_group_pruning_is_credited_before_any_batch_is_read. The other five would also pass a trivial credit-everything-on-close implementation, so those two are the ones doing the work.

push_decoder's existing advance_rg_plan_to_pops_up_to_target and advance_rg_plan_to_errors_when_target_absent are extended to assert that a row group arrow-rs finishes without handing back a reader still credits its bytes.

Also run: the full datafusion-datasource-parquet suite (233 passed), the explain_analyze core tests, the four affected sqllogictest files, cargo fmt, cargo clippy --all-targets --all-features -- -D warnings, and rustdoc with -D warnings.

Are there any user-facing changes?

  • EXPLAIN ANALYZE on a parquet scan gains a bytes_processed= entry, next to bytes_scanned. Four sqllogictest files are updated; the diff there is exactly eight added bytes_processed= fragments and nothing else — existing <slt:ignore> markers and pinned values are preserved.
  • A bullet in docs/source/user-guide/explain-usage.md documenting the metric next to bytes_scanned.
  • No Rust API change: the diff adds and removes no public item. bytes_processed is built on demand rather than stored on ParquetFileMetrics, so the struct is untouched and code constructing it with a struct literal still compiles. Consumers read the counter out of the plan's MetricsSet by name, exactly as they read bytes_scanned — which is the intended way to drive a progress bar or a query watchdog from it.

🤖 Generated with Claude Code

https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3


Generated by Claude Code

`EXPLAIN ANALYZE` reports how many row groups a scan pruned, but nothing
reports how much of a scan is done. `bytes_scanned` looks like it should
serve — its natural denominator, the size of the files in the plan, is
known up front — but it counts only the bytes fetched, so pruning and
projection pushdown leave it understating progress by a factor that
varies per query.

`bytes_processed` completes that numerator: it counts the bytes a scan is
finished with, whether it read them or proved it did not need them. Over
the lifetime of a file it advances by exactly that file's size — or, for
a file split into byte ranges for parallelism, by the size of the range —
so `bytes_processed / total file bytes` is a completion fraction a
progress reporter can use directly.

Credit lands a row group at a time. Row groups pruned while opening the
file (by range, statistics, bloom filter, page index or limit) are
credited before the first batch is decoded, which is the progress a
scan makes that `files_processed` cannot show; row groups dropped
mid-scan by a dynamic filter are credited as they are dropped; the rest
are credited as the decoder reaches them. Whatever is left over is
credited when the file closes, so a scan cut short by a `LIMIT`, an
early stop or an error still ends on exactly the size of its range.

Costs one atomic add per row group. Crediting a row group's bytes
progressively as its rows decode is left to a follow-up.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
@github-actions github-actions Bot added documentation Improvements or additions to documentation core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) datasource Changes to the datasource crate labels Aug 20, 2026
Two fixes from review of the `bytes_processed` metric.

`EarlyStoppingStream` returned `None` once a dynamic filter proved the
rest of a file irrelevant, but went on holding its inner stream. The
decoder's drop is what credits the range's remaining bytes, so the file
kept reading as partly unread after the scan had demonstrably finished
with it — for however long the caller took to drop the wrapper. Hold the
inner stream in an `Option` and release it when marking the stream done,
which also frees the decoder's buffers at the point we stop reading
rather than later. The same applies when the inner stream is simply
exhausted.

`ByteProgress` tracked its remaining budget as a `u64` while crediting a
`Count`, which stores a `usize`. On a 32-bit target a credit above
`usize::MAX` would decrement the budget in full while recording a
saturated value, so the two could disagree and the metric would never
reach the range size. Hold the budget at the counter's width instead, so
an oversized range saturates once, at construction. `bytes_scanned` has
the same ceiling; representing byte counters as `u64` would be a change
to the core metric types, not to this scan.

Both are covered by tests that fail without the corresponding fix.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3

@saadtajwar saadtajwar left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is really nice! Just left a small code-styling suggestion

Comment on lines +1519 to +1534
let mut will_scan = vec![false; rg_metadata.len()];
for entry in &rg_plan {
will_scan[entry.rg_index] = true;
}
let skipped_bytes: u64 = rg_metadata
.iter()
.enumerate()
.filter(|(rg_index, rg_meta)| {
!will_scan[*rg_index]
&& prepared
.file_range
.as_ref()
.is_none_or(|range| row_group_in_range(rg_meta, range))
})
.map(|(_, rg_meta)| row_group_bytes(rg_meta))
.sum();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I think we could make this a bit cleaner by collapsing 1519-1522, suggestion below:

let in_range_bytes: u64 = rg_metadata
    .iter()
    .filter(|rg_meta| {
        prepared
            .file_range
            .as_ref()
            .is_none_or(|range| row_group_in_range(rg_meta, range))
    })
    .map(row_group_bytes)
    .sum();
let skipped_bytes: u64 = in_range_bytes.saturating_sub(rg_plan.iter().map(|e| e.bytes).sum());

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks addressed!

@github-actions github-actions Bot added the auto detected api change Auto detected API change label Aug 20, 2026
claude added 2 commits August 20, 2026 16:16
Review suggestion: the row groups left in `rg_plan` are a subset of the
ones this range owns, because the plan they came from had
`prune_by_range` applied. Subtracting the planned bytes from the range's
total therefore leaves exactly the skipped ones, with no need to build a
lookup of which row groups the plan kept.

Also pin the property the in-range filter exists for: with nothing pruned
a split file's ranges must credit nothing at open, since every row group
a range owns is one it will read and the rest belong to its sibling. The
existing assertions could not catch a range crediting its sibling's row
groups — the drop-time top-up brings the total back to the range size
either way.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
The progress guard was created alongside the decoder, so a file whose
metadata load, filter preparation or bloom-filter load failed never
credited its bytes: the guard that tops up the remainder on drop did not
exist yet. Under `OnError::Skip` the scan carries on past that file, and
`bytes_processed` is left permanently short of the plan's byte total — so
a consumer dividing by it never reaches 100%. `files_processed` already
counts a file that failed to open as processed, and its byte-granular
counterpart should agree.

Move the guard onto the state that travels through the whole open, so
every early return drops it and credits the range. That also folds the
file-level pruning path into the same mechanism: pruning before open now
credits by dropping the guard rather than through a separate `add`, so
there is one way bytes are accounted for rather than two.

Covered by a test that opens a file which is not parquet at all, failing
well before a stream exists; it fails if the guard is built alongside the
decoder as before.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
@adriangb
adriangb requested review from zhuqi-lucas and a balanced review from Copilot August 20, 2026 16:34

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a new Parquet scan completion metric (bytes_processed) that advances monotonically to the file/range effective_size(), enabling reliable scan-progress reporting even with pruning and projection pushdown.

Changes:

  • Introduces ParquetFileMetrics::bytes_processed plus ByteProgress to safely and monotonically account bytes (including on early stop / errors via Drop).
  • Updates Parquet scan/open/decoder logic to credit processed bytes at open-time (for pruned RGs) and during decode (per RG), including split-range correctness via shared row_group_in_range.
  • Updates EXPLAIN ANALYZE outputs, sqllogictest expectations, and user docs to include bytes_processed.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Show a summary per file
File Description
docs/source/user-guide/explain-usage.md Documents the new bytes_processed EXPLAIN metric and its interpretation.
datafusion/sqllogictest/test_files/*.slt Updates expected EXPLAIN ANALYZE metric strings to include bytes_processed.
datafusion/datasource-parquet/src/row_group_filter.rs Extracts row_group_in_range to unify range ownership logic for row groups.
datafusion/datasource-parquet/src/push_decoder.rs Credits bytes_processed as row groups are pruned/decoded/advanced.
datafusion/datasource-parquet/src/opener/mod.rs Creates and wires ByteProgress, credits open-time skipped bytes, and adds invariant tests.
datafusion/datasource-parquet/src/opener/early_stop.rs Ensures inner stream is dropped immediately on early stop to complete byte accounting promptly.
datafusion/datasource-parquet/src/metrics.rs Adds bytes_processed metric and ByteProgress guard implementation.
datafusion/core/tests/sql/explain_analyze.rs Asserts EXPLAIN ANALYZE output contains bytes_processed=.
Suppressed comments (1)

datafusion/datasource-parquet/src/metrics.rs:1

  • Adding a new pub field to a public struct is a semver-breaking change for external users constructing ParquetFileMetrics with a struct literal. If ParquetFileMetrics is intended to be publicly constructible, consider mitigating this by marking the struct #[non_exhaustive] (if not already), or making fields private and exposing construction/access via new()/accessors so future metric additions don’t break downstream builds.
// Licensed to the Apache Software Foundation (ASF) under one

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

/// The on-disk size of a row group, as credited to
/// [`ParquetFileMetrics::bytes_processed`].
fn row_group_bytes(rg_meta: &RowGroupMetaData) -> u64 {
u64::try_from(rg_meta.compressed_size()).unwrap_or(0)
Comment on lines +1533 to +1543
let in_range_bytes: u64 = rg_metadata
.iter()
.filter(|rg_meta| {
prepared
.file_range
.as_ref()
.is_none_or(|range| row_group_in_range(rg_meta, range))
})
.map(row_group_bytes)
.sum();
let planned_bytes: u64 = rg_plan.iter().map(|entry| entry.bytes).sum();
- `DataSourceExec`
- `output_rows=99997497`: A total 99.9M rows were produced
- `bytes_scanned=3703192723`: Of the 14GB file, 3.7GB were actually read (due to projection pushdown)
- `bytes_processed=14779976446`: All 14GB were accounted for: the 3.7GB read, plus the bytes of row groups that pruning ruled out. Comparing this against the total size of the files in the plan tells you how far along a scan is
Adding a `pub` field to `ParquetFileMetrics` is a semver-breaking change
for anyone constructing it with a struct literal, which cargo-semver-checks
reports and reviewers have flagged. The field also turned out to earn
nothing: once file-level pruning started crediting through the progress
guard, the only thing left reading it was the guard's construction.

Build the counter on demand instead, next to the other metrics in this
file that are registered where they are used rather than held on the
struct. The public API is unchanged, so the semver break goes away, and
future metrics of this kind need not widen it either.

`EXPLAIN ANALYZE` output is unaffected: the counter keeps its name, type,
category and filename label, so it renders in the same position. The
sqllogictest expectations are unchanged and still pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
@github-actions github-actions Bot removed the auto detected api change Auto detected API change label Aug 20, 2026
@codecov-commenter

codecov-commenter commented Aug 20, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 98.12332% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.36%. Comparing base (c429919) to head (b8abaaf).
⚠️ Report is 44 commits behind head on main.

Files with missing lines Patch % Lines
datafusion/datasource-parquet/src/push_decoder.rs 86.11% 3 Missing and 2 partials ⚠️
...fusion/datasource-parquet/src/opener/early_stop.rs 98.90% 1 Missing ⚠️
...afusion/datasource-parquet/src/row_group_filter.rs 87.50% 0 Missing and 1 partial ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #24524      +/-   ##
==========================================
+ Coverage   81.24%   81.36%   +0.11%     
==========================================
  Files        1113     1117       +4     
  Lines      392744   397872    +5128     
  Branches   392744   397872    +5128     
==========================================
+ Hits       319090   323717    +4627     
- Misses      54900    55233     +333     
- Partials    18754    18922     +168     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@zhuqi-lucas zhuqi-lucas left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks @adriangb !

Building the counter on demand rather than storing it on
`ParquetFileMetrics` left three intra-doc links pointing at a field that
no longer exists, which fails rustdoc under `-D warnings`.

Repointing them at the struct alone would lose the information, because
there is now no Rust item that carries the metric's contract: consumers
reach it by name through the plan's metrics set, and nothing said so
anywhere rustdoc renders. So the fix is to document it on
`ParquetFileMetrics` itself -- why it is not a field, how to read it, and
what its sum means -- and point the three links there.

Also records the caveat the metric's obvious use invites: it counts work
resolved, not time spent. Pruned bytes are credited as soon as they are
proved unnecessary, and proving that is nearly free, so the ratio is the
right numerator for a progress bar and only a rough one for predicting
how much longer a query will run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
@adriangb
adriangb enabled auto-merge August 21, 2026 15:26
@adriangb
adriangb added this pull request to the merge queue Aug 21, 2026
Merged via the queue into apache:main with commit a6e2d3f Aug 21, 2026
39 checks passed
@adriangb
adriangb deleted the claude/query-progress-tracker-metrics-dyl2u1 branch August 21, 2026 15:45
adriangb added a commit to pydantic/datafusion that referenced this pull request Aug 21, 2026
Backport of apache#24524 (upstream a6e2d3f), carried here so
platform can use it before the fork moves to a DataFusion release that
contains it.

Nothing previously reported how far along a scan was. `bytes_scanned`
counts only the bytes fetched, so pruning and projection pushdown leave it
understating progress by a factor that varies per query; `files_processed`
only moves when a whole file completes, so a scan that opens every file at
once reads zero for the entire first wave however much work is in flight.

`bytes_processed` counts the bytes a scan is finished with, whether it read
them or proved it did not need them. Over the life of a file, or of one
byte range of a file split for parallelism, it advances by exactly that
range's size, so `bytes_processed / total file bytes` is a completion
fraction that needs no second metric to normalise against.

Scope vs upstream: identical, except for the two hunks that adapt
`sync_rg_plan_to_decoder_frontier` and `advance_rg_plan_to`. Those come
from apache#24352, which this branch does not carry, so they are
dropped. `ByteProgress`'s credit-on-drop still holds the metric's
invariant; only the mid-scan crediting of a row group that arrow-rs
finishes without handing back a reader is deferred to file close. Verified
file by file against the upstream squash: every file whose base matched
upstream's is byte-identical after the change, and `push_decoder.rs`
differs only by those dropped hunks.

Tested on this branch: the full `datafusion-datasource-parquet` suite
(172 passed, including the seven `bytes_processed` invariant tests) and
all four sqllogictest files the change touches.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate datasource Changes to the datasource crate documentation Improvements or additions to documentation sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

6 participants